package com.ihr360.binlog.service.impl;

import com.alibaba.otter.canal.protocol.CanalEntry;
import com.ihr360.binlog.component.RosterSalayrChangeRecordEventEmitter;
import com.ihr360.binlog.component.StaffInfoEventEmitter;
import com.ihr360.binlog.dto.CanalConfig;
import com.ihr360.binlog.model.EventRow;
import com.ihr360.binlog.model.SyncTaskOverview;
import com.ihr360.binlog.service.BinlogEventSourceService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import rx.Observable;

import java.util.ArrayList;
import java.util.List;

@Service
public class BinlogEventSourceServiceImpl implements BinlogEventSourceService {
    final static Logger logger = LoggerFactory.getLogger(BinlogEventSourceService.class.getName());
    private boolean allowSync = true;

    private SyncTaskOverview syncTaskOverview = new SyncTaskOverview();

    @Override
    public boolean processEventEntrys(List<CanalEntry.Entry> entrys, CanalConfig config) {
        List<EventRow> rowChanges = new ArrayList<EventRow>();
        for (CanalEntry.Entry entry : entrys) {
            if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                continue;
            }

            CanalEntry.RowChange rowChange = null;
            try {
                rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                logger.info(e.getMessage(), e);
                throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry,
                        e);
            }

            CanalEntry.EventType eventType = rowChange.getEventType();
            logger.debug(String.format(config.getDestination() + "================> binlog[%s:%s] , name[%s,%s] , eventType : %s",
                    entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
                    entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),
                    eventType));
            EventRow syncRow = new EventRow(rowChange, entry.getHeader().getTableName());
            rowChanges.add(syncRow);
        }
        if (!CollectionUtils.isEmpty(rowChanges)) {
            for (EventRow syncRow : rowChanges) {
                publishEvent(syncRow);
            }
        }
        return true;
    }

    @Autowired
    StaffInfoEventEmitter staffInfoEventEmitter;

    @Autowired
    RosterSalayrChangeRecordEventEmitter rosterSalayrChangeRecordEventEmitter;

    private void publishEvent(EventRow eventRow) {
        Observable.just(eventRow).filter(row ->
                row.getTableName().equals("tab_staff_info")).subscribe(staffInfoEventEmitter.onEventRow);

        Observable.just(eventRow).filter(row ->
                row.getTableName().equals("tab_roster_salary_change_record")).subscribe(rosterSalayrChangeRecordEventEmitter.onEventRow);

    }

    @Override
    public SyncTaskOverview syncStatus() {
        return syncTaskOverview;
    }


    @Override
    public SyncTaskOverview fullSync(String schema, String tableName, String companyId) {
        //是否考虑使用ETL工具完成全量同步
        return null;
    }

    @Override
    public SyncTaskOverview syncUp() {
        allowSync = true;
        syncTaskOverview.message = "Sync is up now ";
        return syncTaskOverview;
    }

    @Override
    public SyncTaskOverview syncDown() {
        allowSync = false;
        syncTaskOverview.message = "Sync is down now ";
        return syncTaskOverview;
    }
}